package com.parsechina.hpush.broker.internal;

import com.parsechina.hpush.broker.internal.subscriptions.Subscription;
import com.parsechina.hpush.broker.internal.subscriptions.SubscriptionsStore;
import com.parsechina.hpush.broker.parser.PublishMessage;
import com.parsechina.hpush.broker.parser.PushMessage;
import com.parsechina.hpush.broker.spi.IMessagesStore;
import com.parsechina.hpush.broker.spi.ISessionsStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * Model a Session like describe on page 25 of MQTT 3.1.1 specification:
 * The Session state in the Server consists of:
 * <ul>
 * <li>The existence of a Session, even if the rest of the Session state is empty.</li>
 * <li>The Client’s subscriptions.</li>
 * <li>QoS 1 and QoS 2 messages which have been sent to the Client, but have not been
 * completely acknowledged.</li>
 * <li>QoS 1 and QoS 2 messages pending transmission to the Client.</li>
 * <li>QoS 2 messages which have been received from the Client, but have not been
 * completely acknowledged.</li>
 * <li>Optionally, QoS 0 messages pending transmission to the Client.</li>
 * </ul>
 *
 * @author andrea
 */
public class ClientSession {

    private static final Logger LOG = LoggerFactory.getLogger(ClientSession.class);

    public final String clientId;

    private final IMessagesStore messagesStore;

    private final ISessionsStore m_sessionsStore;

    private final Set<Subscription> subscriptions = new HashSet<>();

    private volatile boolean cleanSession;

    private final Queue<PushMessage> m_queueToPublish = new ConcurrentLinkedQueue<PushMessage>();

    public ClientSession(String clientId, IMessagesStore messagesStore, ISessionsStore sessionsStore,
                         boolean cleanSession) {
        this.clientId = clientId;
        this.messagesStore = messagesStore;
        m_sessionsStore = sessionsStore;
        this.cleanSession = cleanSession;
    }

    /**
     * @return the list of messages to be delivered for client related to the session.
     */
    public List<StoredMessage> storedMessages() {
        //read all messages from enqueued store
        Collection<MessageGUID> guids = m_sessionsStore.enqueued(clientId);
        return messagesStore.listMessagesInSession(guids);
    }

    /**
     * Remove the messages stored in a cleanSession false.
     */
    public void removeEnqueued(int messageId, MessageGUID guid) {
        m_sessionsStore.removeEnqueued(clientId, messageId, guid);
    }

    /**
     * Remove the messages stored in a cleanSession false.
     */
    public void removeMessageGuid(int messageId, MessageGUID guid) {
        m_sessionsStore.removeMessageGUID(clientId, messageId, guid);
    }

    @Override
    public String toString() {
        return "ClientSession{clientId='" + clientId + '\'' + "}";
    }

    public boolean subscribe(Subscription newSubscription) {
        LOG.info("<{}> subscribed to the topic filter <{}> with QoS {}",
                newSubscription.getClientId(), newSubscription.getTopicFilter(),
                newSubscription.getRequestedQos());

        boolean validTopic = SubscriptionsStore.validate(newSubscription.getTopicFilter());
        if (!validTopic) {
            //send SUBACK with 0x80 for this topic filter
            return false;
        }
        ISessionsStore.ClientTopicRelation matchingCouple = new ISessionsStore.ClientTopicRelation(clientId, newSubscription.getTopicFilter());
        Subscription existingSub = m_sessionsStore.getSubscription(matchingCouple);
        //服务质量只有比存在的大才会更新订阅
//        if (existingSub == null || existingSub.getRequestedQos().value() < newSubscription.getRequestedQos().value()) {
//            if (existingSub != null) {
//                subscriptions.remove(newSubscription);
//            }
//            subscriptions.add(newSubscription);
//            m_sessionsStore.addNewSubscription(newSubscription);
//        }

        if (existingSub != null) {
            subscriptions.remove(newSubscription);
        }
        subscriptions.add(newSubscription);
        m_sessionsStore.addNewSubscription(newSubscription);
        return true;
    }

    public void unsubscribeFrom(String topicFilter) {
        m_sessionsStore.removeSubscription(topicFilter, clientId);
        Set<Subscription> subscriptionsToRemove = new HashSet<>();
        for (Subscription sub : subscriptions) {
            if (sub.getTopicFilter().equals(topicFilter)) {
                subscriptionsToRemove.add(sub);
            }
        }
        subscriptions.removeAll(subscriptionsToRemove);
    }

    public void disconnect() {
        if (cleanSession) {
            //cleanup topic subscriptions
            cleanSession();
        }

    }

    public void cleanSession() {
        LOG.info("cleaning old saved subscriptions for client <{}>", clientId);
        m_sessionsStore.removeAllSubscriptions(clientId);
        LOG.debug("Wiped subscriptions for client <{}>", clientId);

        //remove also the messages stored of type QoS1/2
        messagesStore.dropMessagesInSession(clientId);
        LOG.debug("Removed messages in session for client <{}>", clientId);
    }

    public boolean isCleanSession() {
        return cleanSession;
    }

    public void cleanSession(boolean cleanSession) {
        this.cleanSession = cleanSession;
        m_sessionsStore.updateCleanStatus(clientId, cleanSession);
    }

    public int nextPacketId() {
        return m_sessionsStore.nextPacketId(clientId);
    }

    public MessageGUID inFlightAcknowledged(int messageId) {
        LOG.trace("Acknowledging inflight, clientId <{}> messageID {}", clientId, messageId);
        return m_sessionsStore.inFlightAck(clientId, messageId);
    }

    public void inFlightAckWaiting(MessageGUID guid, int messageId) {
        LOG.trace("Adding to inflight {}, guid <{}>", messageId, guid);
        m_sessionsStore.inFlight(clientId, messageId, guid);
    }

    public StoredMessage secondPhaseAcknowledged(int messageId) {
        MessageGUID guid = m_sessionsStore.secondPhaseAcknowledged(clientId, messageId);
        if (guid == null) {
            return null;
        }
        return messagesStore.getMessageByGuid(guid);
    }

    public void enqueueToDeliver(MessageGUID guid) {
        m_sessionsStore.bindToDeliver(guid, clientId);
    }

    public StoredMessage storedMessage(int messageId) {
        final MessageGUID guid = m_sessionsStore.mapToGuid(clientId, messageId);
        if (guid == null) {
            return null;
        }
        return messagesStore.getMessageByGuid(guid);
    }

    /**
     * Enqueue a message to be sent to the client.
     *
     * @return false if the queue is full.
     */
    public boolean enqueue(PublishMessage pubMessage) {
        return m_queueToPublish.offer(pubMessage);
    }

    public PushMessage dequeue() {
        return m_queueToPublish.poll();
    }

    public void moveInFlightToSecondPhaseAckWaiting(int messageId) {
        m_sessionsStore.moveInFlightToSecondPhaseAckWaiting(clientId, messageId);
    }

    public StoredMessage getInflightMessage(int messageId) {
        return m_sessionsStore.getInflightMessage(clientId, messageId);
    }
}
